package com.eshipenet.shipnet.dp_decompressor

import java.io.ByteArrayInputStream
import java.nio.ByteBuffer

import com.sksamuel.avro4s.AvroSchema
import org.apache.avro.{Schema, SchemaBuilder}
import org.apache.avro.io.DecoderFactory

//import com.sksamuel.avro4s._
import org.apache.avro.specific.SpecificDatumReader
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.scala._
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema
import org.apache.kafka.clients.consumer.ConsumerRecord

//object EdgeChannelDataPacketKafkaDeserializer {
//  val msgKeySchema: Schema = AvroSchema[MsgKey]
//}

case class ValueWrapper(value: Array[Byte])

object EdgeChannelDataPacketKafkaDeserializer {
  lazy val valueSchema: Schema = AvroSchema[ValueWrapper]
}

class EdgeChannelDataPacketKafkaDeserializer extends KafkaDeserializationSchema[EdgeChannelDataPacketKafkaMessage] {

  private lazy val keyDecoder   = DecoderFactory.get()
  private lazy val valueDecoder = DecoderFactory.get()
  //  private val valueSchema = AvroSchema[ValueWrapper]

  override def isEndOfStream(nextElement: EdgeChannelDataPacketKafkaMessage): Boolean =
    false

  override def deserialize(record: ConsumerRecord[Array[Byte], Array[Byte]]): EdgeChannelDataPacketKafkaMessage = {
    val key   = parseKey(record.key())
    val value = parseValue(record.value())
    EdgeChannelDataPacketKafkaMessage(key.topic, value)

  }

  private def parseKey(buf: Array[Byte]): MsgKey = {
    val stream  = new ByteArrayInputStream(buf.slice(5, buf.length))
    val decoder = keyDecoder.directBinaryDecoder(stream, null)
    val reader  = new SpecificDatumReader[MsgKey](MsgKey.SCHEMA$)
    reader.read(null, decoder)
  }

  private def parseValue(bytes: Array[Byte]): String = {
    val decoder = valueDecoder.binaryDecoder(bytes.slice(5, bytes.length), null)
    val reader =
      new SpecificDatumReader[ByteBuffer](SchemaBuilder.builder().bytesType())
    val b = reader.read(null, decoder)
    new String(b.array(), "UTF-8")
  }

  //  private def parseValue(bytes: Array[Byte]): String = {
  //    println(s"buf = ${bytes.map(b => b.toHexString).mkString(",")}")
  //    val s = new String(bytes, "UTF-8")
  //    s.split("\\{", 2) match {
  //      case Array(_, s2) => s2
  //      case _            => throw new Exception(s"invalid payload: $s")
  //    }
  //  }

  override def getProducedType: TypeInformation[EdgeChannelDataPacketKafkaMessage] =
    createTypeInformation[EdgeChannelDataPacketKafkaMessage]

}
